/*
 * MIT License
 *
 * Copyright (c) 2020 wen.gu <454727014@qq.com>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

 /***************************************************************************
 * Name: cbus_srv_event.c
 *
 * Purpose: implementation event publisher for cbus service
 *
 * Developer:
 *   wen.gu , 2022-02-16
 *
 * TODO:
 *
 ***************************************************************************/

 /******************************************************************************
 **    INCLUDES
 ******************************************************************************/

#include "cbus/bus/cbus_event_publisher.h"

#include <string.h>
#include <stdlib.h>

#include "cbus/core/cbus_map.h"

#include "cbus/core/cbus_log.h"
/******************************************************************************
 **    MACROS
 ******************************************************************************/

/******************************************************************************
 **    VARIABLE DEFINITIONS
 ******************************************************************************/
typedef struct _event_subscribe_node
{
    struct _event_subscribe_node* next;
    cbus_endpoint_id endpoint_id;
    cbus_endpoint_t endpoint_ref;
}event_subscribe_node_t;

typedef struct _event_subscribe_info
{
    event_subscribe_node_t* subscribers; /** who(which client) subscribe current event */
    cbus_message_t msg;
}event_subscribe_info_t;

struct _endpoint_info
{
    cbus_endpoint_id endpoint_id; /** indicat a id for remote endpoint */
    cbus_endpoint_t endpoint_ref; /** indicat local endpoint */
};

struct _cbus_event_publisher
{
    /** this strcut will not be used, just as a adaptor */
    void* dummy;
};

/******************************************************************************
 **    inner FUNCTION DEFINITIONS
 ******************************************************************************/
static void event_subscribe_node_free(event_subscribe_info_t* info)
{
    event_subscribe_node_t* item = info->subscribers;
    while (item)
    {
        event_subscribe_node_t* next = item->next;
        free(item);
        item = next;
    }   
    info->subscribers = NULL; 
}

////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////

static event_subscribe_info_t* event_subscribe_info_alloc()
{
    event_subscribe_info_t* info = (event_subscribe_info_t*)malloc(sizeof(event_subscribe_info_t));

    if (info)
    {
        memset(info, 0, sizeof(event_subscribe_info_t));
    }

    return info;
}

static GErrc event_subscribe_info_add(event_subscribe_info_t* subscribe_info, cbus_endpoint_t endpoint_ref, cbus_endpoint_id endpoint_id)
{
    event_subscribe_node_t* item = (event_subscribe_node_t*)malloc(sizeof(event_subscribe_node_t));

    if (!item)
    {
        LOGE("alloc client node for event subscribe failed(endpoint_ref: %p, endpoint_id: %p)\n", endpoint_ref, endpoint_id);
        return G_ErrInsufficientResources;
    }

    item->endpoint_id  = endpoint_id;
    item->endpoint_ref = endpoint_ref;
    item->next = subscribe_info->subscribers;
    subscribe_info->subscribers = item;

    return G_OK;
}

static GErrc event_subscribe_info_del(event_subscribe_info_t* subscribe_info, cbus_endpoint_t endpoint_ref, cbus_endpoint_id endpoint_id)
{
    event_subscribe_node_t* item = subscribe_info->subscribers;
    event_subscribe_node_t* prev = item;
    while (item)
    {
        if ((endpoint_ref == item->endpoint_ref) && (item->endpoint_id == endpoint_id))
        {
            if (prev != item)    
            {
                prev->next = item->next;
            }
            else
            {/** indicate head node */
                subscribe_info->subscribers = item->next;
            }
            
            free(item);
            return G_OK;
        }

        prev = item;
        item = item->next;
    }
    
    return G_ErrNotFound;
}

static void event_subscribe_info_free(void* ptr)
{
    event_subscribe_info_t* info = (event_subscribe_info_t*)ptr;
    event_subscribe_node_free(info);
    cbus_message_free(info->msg);
    free(info);
}

static void event_on_remove(GPTR opaque, GU64 key, GU32 type, const void* value, GU32 value_size)
{
    event_subscribe_info_t* info = (event_subscribe_info_t*)value;

    struct _endpoint_info* ep_info = (struct _endpoint_info*)opaque;

    GErrc ret = event_subscribe_info_del(info, ep_info->endpoint_ref, ep_info->endpoint_id);
    if (G_OK != ret)
    {
        LOGW("remove subscribe for event(%llu) for endpoint ref(%p) and endpoint id: (%p) failed(0x%08x)\n",
            key,ep_info->endpoint_ref, ep_info->endpoint_id, ret);
    }    
}

/******************************************************************************
 **    FUNCTION DEFINITIONS
 ******************************************************************************/
cbus_event_publisher_t cbus_event_publisher_create()
{
    /** here is a map , like
     * map<message_id, event_subscribe_info_t*>
     * 
     */
    return (cbus_event_publisher_t)cbus_map_create(CMV_TYPE_Pointer, CBUS_MAP_DEFAULT_CAPACITY);
}

void cbus_event_publisher_destroy(cbus_event_publisher_t publisher)
{
    if (publisher)
    {
        cbus_map_destroy((cbus_map_t)publisher);
    }
}

/** clear all subscriber info */
GErrc cbus_event_publisher_clear_subsciber(cbus_event_publisher_t publisher)
{
    if (!publisher)
    {
        return G_ErrBadParameter;
    }

    cbus_map_clear((cbus_map_t)publisher);
    return G_OK;
}

GErrc cbus_event_publisher_remove_subscriber(cbus_event_publisher_t publisher, cbus_endpoint_t endpoint_ref, cbus_endpoint_id endpoint_id)
{
    if (!publisher || !endpoint_ref || !endpoint_id)
    {
        return G_ErrBadParameter;
    }
    struct _endpoint_info ep_info;
    ep_info.endpoint_ref = endpoint_ref;
    ep_info.endpoint_id  = endpoint_id;

    return (cbus_map_for_each((cbus_map_t)publisher, event_on_remove, &ep_info) == GTRUE) ? G_OK : G_ErrUndefined;
}

void cbus_event_publisher_on_stop_subscribe(cbus_message_t stop_subscribe_message)
{
    cbus_map_t event_map = (cbus_map_t)cbus_message_get_opaque(stop_subscribe_message);
    event_subscribe_info_t* subscribe_info = NULL;

    if (cbus_map_find_pointer(event_map, cbus_message_get_message_id(stop_subscribe_message), (const void**)&subscribe_info) == GTRUE)
    {
        GErrc ret = event_subscribe_info_del(subscribe_info, cbus_message_get_endpoint_ref(stop_subscribe_message), cbus_message_get_endpoint_id(stop_subscribe_message));
        if (G_OK == ret)
        {
            cbus_message_send_response_code(stop_subscribe_message, CBUS_MSG_OK);
        }
        else
        {
            LOGW("remove subscribe for event(%llu) for endpoint ref(%p) and endpoint id: (%p) failed(0x%08x)\n",
                cbus_message_get_message_id(stop_subscribe_message), cbus_message_get_endpoint_ref(stop_subscribe_message), 
                cbus_message_get_endpoint_id(stop_subscribe_message), ret);
            cbus_message_send_response_code(stop_subscribe_message, cbus_gerrc_to_msg_return_code(ret));
        }         
    }  
    else
    {
        cbus_message_send_response_code(stop_subscribe_message, CBUS_MSG_ErrNotFound);
    }
}

void cbus_event_publisher_on_subscribe(cbus_message_t subscribe_message)
{
    cbus_map_t event_map = (cbus_map_t)cbus_message_get_opaque(subscribe_message);
    event_subscribe_info_t* subscribe_info = NULL;

    if (cbus_map_find_pointer(event_map, cbus_message_get_message_id(subscribe_message), (const void**)&subscribe_info) == GFALSE)
    {
        subscribe_info = event_subscribe_info_alloc();
        if (!subscribe_info)
        {
            cbus_message_send_response_code(subscribe_message, CBUS_MSG_ErrInsufficientResources);
            return ;
        }

        if (cbus_map_set_pointer(event_map, cbus_message_get_message_id(subscribe_message), subscribe_info, event_subscribe_info_free) == GFALSE)
        {
            event_subscribe_info_free(subscribe_info);
            cbus_message_send_response_code(subscribe_message, CBUS_MSG_ErrInsufficientResources);
            return ;
        }
    }

    GErrc ret = event_subscribe_info_add(subscribe_info, cbus_message_get_endpoint_ref(subscribe_message), cbus_message_get_endpoint_id(subscribe_message));
    
    cbus_message_send_response_code(subscribe_message, cbus_gerrc_to_msg_return_code(ret));
}

GErrc cbus_event_publisher_notify(cbus_event_publisher_t publisher, cbus_message_t event)
{
    if (!publisher || !event)
    {
        return G_ErrBadParameter;
    }

    cbus_map_t event_map = (cbus_map_t)publisher;
    event_subscribe_info_t* subscribe_info = NULL;

    if (cbus_map_find_pointer(event_map, cbus_message_get_message_id(event), (const void**)&subscribe_info) == GTRUE)
    {
        if (subscribe_info->subscribers)
        {
            cbus_session_id session_id = 1;
            if (subscribe_info->msg)
            {
                session_id = cbus_message_get_session_id(subscribe_info->msg) + 1;
                cbus_message_free(subscribe_info->msg);
            }
            subscribe_info->msg = cbus_message_reference(event);

            if (!subscribe_info->msg)
            {
                G_ErrUndefined;
            }

            cbus_message_t msg_ref = subscribe_info->msg;
            cbus_message_set_session_id(msg_ref, session_id);

            event_subscribe_node_t* item = subscribe_info->subscribers;
            while (item)
            {
                cbus_message_set_endpoint_id(msg_ref, item->endpoint_id);
                cbus_message_set_endpoint_ref(msg_ref, item->endpoint_ref);
                //cbus_service_message_send(publisher->service_ref, msg_ref, GFALSE); /** todo, process error */
                cbus_message_send(msg_ref);
                item = item->next;
            }  
            return G_OK;
        }
    }
    
    return G_ErrNoSubscriber;
}